package network;

import java.io.BufferedOutputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.io.PrintWriter;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.protobuf.ByteString;

import command.protobuf.MessageBufProtoBuf.RpcMessage;
import config.Configure;


/**
 * OutputConnectionPool is responsible for manage outputconnection; This
 * connection can only be retrieved from OutputConnectionPool. When
 * OutputConnectionPool finds this connection has been idle from some time. it
 * removes this outputconnection from the pool and send a CLOSE_MESSAGE to cause
 * this connection to be closed.
 * 
 * @author xiafan
 * 
 */
public class OutputTcpConnection extends Thread {
	public static Logger logger = LoggerFactory
			.getLogger(OutputTcpConnection.class);
	public static final RpcMessage CLOSE_MESSAGE = makeCloseMessage();

	private InetAddress target = null;
	private Integer port = -1;
	private Socket socket = null;
	private DataOutputStream output = null;
	private BlockingQueue<RpcMessage> queue = new LinkedBlockingQueue<RpcMessage>();
	private long lastCommTime = System.currentTimeMillis();
	
	public OutputTcpConnection(Socket socket) {
		this.socket = socket;
	}

	public OutputTcpConnection(InetAddress target) {
		this(target, Configure.getListenPort());
	}

	public OutputTcpConnection(InetAddress target, Integer port) {
		this.target = target;
		this.port = port;
	}

	public void write(RpcMessage message) {
		try {
			queue.put(message);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}

	@Override
	public void run() {
		while (true) {
			try {
				RpcMessage message = queue.take();
				lastCommTime = System.currentTimeMillis();

				if (message == CLOSE_MESSAGE) {
					close();
					return;
				}
				if ((output != null && socket.isConnected()) || connect()) {
					// message.writeTo(socket.getOutputStream());
					message.writeTo(output);
					output.flush();
				} else {
					close();
					return;
				}
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
	}

	public long getIdleTimeInMillis() {
		return System.currentTimeMillis() - lastCommTime;
	}

	public boolean connect() {
		logger.debug("output connection connect " + target + " " + port);
		if (socket != null && socket.isConnected() && output != null)
			return true;
		try {
			if (socket == null || !socket.isConnected()) {
				socket = new Socket(target, port);
				//socket.setKeepAlive(true);
				socket.setTcpNoDelay(true);
				// SocketAddress sockAddr = new InetSocketAddress(target, port);
				// socket.connect(sockAddr);
				if (socket == null)
					return false;
			}
			output = new DataOutputStream(new BufferedOutputStream(
					socket.getOutputStream(), 4096));
		} catch (IOException e) {
			socket = null;
			output = null;
			e.printStackTrace();
			return false;
		}
		return true;
	}

	public void close() {
		logger.debug("close output connection: " + socket.toString());
		queue.clear();
		try {
			//socket.close();
			socket.shutdownOutput();
		} catch (IOException e) {
			e.printStackTrace();
		}
		socket = null;
		output = null;
	}

	public Socket getSocket() {
		return socket;
	}
	
	private static RpcMessage makeCloseMessage() {
		RpcMessage.Header header = RpcMessage.Header.newBuilder().setFlag(0).setCmdType(0).setID(-1).build();
		return RpcMessage.newBuilder().setHeader(header).setContent(ByteString.EMPTY).build();
	}
}
